package com.es;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.search.Hit;
import com.es.param.EsAggsParam;
import com.es.param.EsSearchPageParam;
import com.es.param.EsSearchParam;
import com.es.param.EsRestDocParam;
import com.es.vo.EsRestBulkDocVo;
import com.es.vo.EsRestIndexVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * 基于 elasticsearch-rest-client 8.1.0 开发工具类
 * - jdk > 1.8
 * - es > 8.x
 * @author heyonghao
 * @date 2023/2/3
 */
@Slf4j
@Component
public class EsRestClientUtil {
    private static final String ES_FAIL_CONNECTION="es服务异常";

    private final ElasticsearchClient elasticsearchClient;
    @Autowired
    public EsRestClientUtil(ElasticsearchClient elasticsearchClient) {
        this.elasticsearchClient = elasticsearchClient;
    }

    /**
     * 判断索引是否存在
     * @param indexName 索引名称
     * @return Boolean
     */
    public Boolean existIndex(String indexName){
        try {
            return elasticsearchClient.indices().exists(e->e.index(indexName)).value();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return Boolean.FALSE;
        }
    }

    /**
     * 创建索引
     * Property.of(e->e)：不配置对应 setting
     * demo:
     *        EsRestIndexVo esRestIndexVo = new EsRestIndexVo();
     *         esRestIndexVo.setName("goos_index");
     *         esRestIndexVo.setAliases("goods");
     *         esRestIndexVo.setAliasSetting(Alias.of(e->e.searchRouting("1,2")));
     *         Map<String, Property> property = new HashMap<>(16);
     *         property.put("goods_name",Property.of(p->p.text(e->e.analyzer("ik_max_word"))));
     *         property.put("price",Property.of(p->p.double_(e->e.nullValue(0d))));
     *         property.put("stock",Property.of(p->p.integer(e->e.nullValue(0))));
     *         property.put("desc",Property.of(p->p.text(e->e.analyzer("ik_max_word"))));
     *         property.put("tags",Property.of(p->p.keyword(e->e.nullValue(""))));
     *         property.put("in_stock_time",Property.of(p->p.date(e->e.format("yyyy-MM-dd HH:mm:ss"))));
     *         esRestIndexVo.setProperty(property);
     * @param esRestIndexVo 索引创建vo
     * @return Boolean
     */
    public Boolean createIndex(EsRestIndexVo esRestIndexVo){
        try {
            return elasticsearchClient.indices().create(e -> e.index(esRestIndexVo.getName())
                    .aliases(esRestIndexVo.getAliases(), esRestIndexVo.getAliasSetting())
                    .settings(set -> set.numberOfReplicas(esRestIndexVo.getNumberOfReplicas()).numberOfShards(esRestIndexVo.getNumberOfShards()))
                    .mappings(s -> s.properties(esRestIndexVo.getProperty()))
            ).acknowledged();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return Boolean.FALSE;
        }
    }

    /**
     * 删除索引
     * @param indexName 索引名称
     * @return  Boolean
     */
    public Boolean delIndex(String indexName){
        try {
            return elasticsearchClient.indices().delete(e->e.index(indexName)).acknowledged();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return Boolean.FALSE;
        }
    }

    /**
     * 创建索引别名
     * @param indexName 索引
     * @param aliases   别名
     * @return Boolean
     */
    public Boolean addIndexAliases(String indexName,String aliases){
        try {
            return elasticsearchClient.indices().putAlias(e->e.index(indexName).name(aliases)).acknowledged();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return Boolean.FALSE;
        }
    }

    /**
     * 删除索引别名
     * @param indexName 索引
     * @param aliases   别名
     * @return Boolean
     */
    public Boolean delIndexAliases(String indexName, List<String> aliases){
        try {
            return elasticsearchClient.indices().deleteAlias(e->e.index(indexName).name(aliases)).acknowledged();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return Boolean.FALSE;
        }
    }

    /**
     * 索引库文档数量
     * @param indexName 索引名称
     * @return Long-文档总数
     */
    public Long countIndex(String indexName){
        try {
            return elasticsearchClient.count(e->e.index(indexName)).count();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return null;
        }
    }

    /**
     * 添加文档
     * @param indexName 索引名称
     * @param id  文档id
     * @param doc 文档数据
     * @return id
     */
    public String addDoc(String indexName,String id,Object doc){
        try {
            return elasticsearchClient.create(e->e.index(indexName).id(id).document(doc)).id();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return null;
        }
    }

    /**
     * 10w数据在10秒左右，毕竟es强大在搜索
     * 批量添加索引文档
     * @param indexName 索引名称
     * @param docs      文档集合
     * @return 消耗时间
     */
    public Long addBatchDoc(String indexName,List<EsRestBulkDocVo> docs){
        if (CollectionUtil.isEmpty(docs)){
            return null;
        }
        System.out.println("赋值id开始"+ DateUtil.now());
        docs.forEach(e->e.setDocId(String.valueOf(IdUtil.getSnowflakeNextId())));
        System.out.println("赋值id结束"+ DateUtil.now());
        List<BulkOperation> bulkOperations = new CopyOnWriteArrayList<>();
        System.out.println("封装请求对象开始"+ DateUtil.now());
        docs.forEach(e->{
            bulkOperations.add(new BulkOperation.Builder().create(d-> d.document(e.getDoc()).id(e.getDocId()).index(e.getIndexName())).build());
        });
        System.out.println("封装请求对象结束"+ DateUtil.now());
        try {
            return elasticsearchClient.bulk(e->e.index(indexName).operations(bulkOperations)).took();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return null;
        }
    }

    /**
     * 更新文档
     * @param indexName 索引名称
     * @param id  文档id
     * @param doc 文档数据
     * @return id
     */
    public String editDocById(String indexName,String id,Object doc,Class c){
        try {
            return elasticsearchClient.update(u -> u.index(indexName).id(id).doc(doc), c).id();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return null;
        }
    }

    /**
     * 根据文档id，查询对应文档
     * @param param 索引文档查询param
     * @param <T>   文档对象
     * @return      source
     */
    public <T> T getDocById(EsRestDocParam<T> param){
        try {
            return elasticsearchClient.get(g -> g
                            .index(param.getIndexName())
                            .id(param.getDocId())
                    , param.getDocClass()
            ).source();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return null;
        }
    }

    /**
     * 文档单条件匹配查询
     * List<GoodsVo> list = hits.stream().map(Hit::source).collect(Collectors.toList());
     * filed: keyword 全匹配查询
     * filed: text  分词匹配查询
     * @param param 查询;默认最多查询 10000条数据，没有深分页得情况
     * @param <T> 返回
     * @return  List<Hit<T>>
     */
    public <T> List<Hit<T>> listMatchOne(EsSearchParam<T> param){
        try {
            return elasticsearchClient.search(e ->
                            e.index(param.getIndexName())
                                    .query(q -> q.match(m -> m.field(param.getFiled()).query(param.getValue())))
                                    .sort(s->s.field(f->f.field(param.getSortFiled()).order(param.getSortOrder())))
                    , param.getTargetClass()).hits().hits();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return Collections.emptyList();
        }
    }

    /**
     * 匹配分页查询; 默认只能查询 10000条之前的数据
     * @param param 查询参数
     * @param <T>   目标对象
     * @return List<Hit<T>>
     */
    public <T> List<Hit<T>> listMatchPage(EsSearchPageParam<T> param){
        Long countIndex = countIndex(param.getIndexName());
        if (countIndex>10_0000){
            int curSearchCount = param.getPage() * param.getSize();
            if (curSearchCount>10_0000){
                log.info("分页搜索限制，默认from、size分页查询深度不超过1w条");
                return Collections.emptyList();
            }
        }
        int from = (param.getPage() - 1) * param.getSize();
        int size = param.getSize();

        try {
            return elasticsearchClient.search(e ->
                            e.index(param.getIndexName())
                                    .query(q -> q.match(m -> m.field(param.getFiled()).query(param.getValue())))
                                    .sort(s->s.field(f->f.field(param.getSortFiled()).order(param.getSortOrder())))
                                    .from(from)
                                    .size(size)
                    , param.getTargetClass()).hits().hits();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return Collections.emptyList();
        } catch (Exception e){
            e.printStackTrace();
            return Collections.emptyList();
        }
    }


    /**
     * 基本分组计算查询
     * @param param param
     * @return Map<String, Aggregate>
     */
    public Map<String, Aggregate> aggsByFiled(EsAggsParam param){
        try {
            Map<String, Aggregate> aggregations = elasticsearchClient.search(e ->
                            e.index(param.getIndexName())
                                    .aggregations(param.getGroupName(), a -> a.terms(t->t.field("tags")))
                    , param.getTargetClass()).aggregations();
            return aggregations;
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return null;
        }
    }

    /**
     * 范围查询
     * @param param
     * @param <T>
     * @return
     */
    public <T> List<Hit<T>> listRange(EsSearchParam<T> param){
        try {
            return elasticsearchClient.search(e ->
                            e.index(param.getIndexName())
                                    .query(q -> q.range(r->r.field(param.getFiled()).gt(null)))
                                    .sort(s->s.field(f->f.field(param.getSortFiled()).order(param.getSortOrder())))
                    , param.getTargetClass()).hits().hits();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return Collections.emptyList();
        }
    }


    /**
     * 匹配查询-> 高亮字段
     * @param param param
     * @param <T> 目标对象
     * @return  List<Hit<T>>
     */
    public <T> List<Hit<T>> listHighlight(EsSearchParam<T> param){
        try {
            return elasticsearchClient.search(e ->
                            e.index(param.getIndexName())
                                    .query(q -> q.match(m -> m.field(param.getFiled()).query(param.getValue())))
                                    .highlight(h->h.fields(param.getHlightFiled(),f->f.preTags("<font color='red'>").postTags("</font>")))
                                    .sort(s->s.field(f->f.field(param.getSortFiled()).order(param.getSortOrder())))
                    , param.getTargetClass()).hits().hits();
        } catch (IOException e) {
            log.info(ES_FAIL_CONNECTION);
            return Collections.emptyList();
        }
    }




}
